package com.atguigu.app.dwm;

import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.DWDStaticConstants;
import com.atguigu.common.DWMStaticConstants;
import com.atguigu.utils.KafkaUtil;
import org.apache.flink.api.common.functions.RichFilterFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @Author:GaoFei
 * @Description:
 * @Date:Created in 18:11
 * @Modified By:
 *
 * {"common":{"ar":"530000","ba":"vivo","ch":"xiaomi","is_new":"0","md":"vivo iqoo3",
 * "mid":"mid_8","os":"Android 11.0","uid":"33","vc":"v2.1.134"},
 * "start":{"entry":"install","loading_time":11304,"open_ad_id":12,"open_ad_ms":2078,"open_ad_skip_ms":0},
 * "ts":1608276470000}
 */
public class UniqueVisitApp {
    private static Logger logger = LoggerFactory.getLogger("uvLog");

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        /**
         * 生产环境下一定要写的，不过学习阶段还要开启hdfs比较麻烦先关闭
         *         System.setProperty("HADOOP_USER_NAME","atguigu");
         *         env.setStateBackend(new FsStateBackend(OdsStaticConstants.CHECK_POINT_URL));
         *         // 设置checkpoint时间
         *         env.enableCheckpointing(5000L);
         *         // 设置精准一次性
         *         env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
         *         // 设置check超时时间
         *         env.getCheckpointConfig().setCheckpointTimeout(10000L);
         *         // 设置最多两个checkpoint一块
         *         env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
         *         // 设置两个check间隔
         *         env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
         */
        // 消费Kafka ods_base_db主题数据创建流
        DataStreamSource<String> dataStreamSource =
                env.addSource(KafkaUtil.getKafkaConsumer(DWMStaticConstants.DWM_SOURCE_TOPIC, DWMStaticConstants.DWM_PAGE_LOG_GROUP_ID));
        // 将每行数据转换为JSON对象
        SingleOutputStreamOperator<JSONObject> midFilterByDay = dataStreamSource
                .map(JSONObject::parseObject)
                .keyBy(f -> f.getJSONObject("common").getString("mid"))
                .filter(new RichFilterFunction<JSONObject>() {
                    private ValueState<String> dateState;
                    private SimpleDateFormat simpleDateFormat;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> stateDescriptor = new ValueStateDescriptor<>("date-state", String.class);
                        // 给状态设置过期时间 24H
                        StateTtlConfig build = StateTtlConfig.newBuilder(Time.hours(24))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .build();
                        stateDescriptor.enableTimeToLive(build);
                        dateState = getRuntimeContext().getState(stateDescriptor);
                        simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
                    }

                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        String last_page_id = value.getJSONObject("page").getString("last_page_id");
                        if (last_page_id == null || last_page_id.length() <= 0) {
                            String currentDay = simpleDateFormat.format(value.getLong("ts"));
                            if (!currentDay.equals(dateState.value())) {
                                dateState.update(currentDay);
                                return true;
                            }
                        }
                        return false;
                    }
                });

        midFilterByDay
                .map(JSONAware::toJSONString)
                .addSink(KafkaUtil.getKafkaProducter(DWMStaticConstants.DWM_PAGE_LOG_SINK_TABLE));
        logger.info("uv过滤去重完成", midFilterByDay);
        midFilterByDay.print(">>>>>>");
        env.execute();
    }
}
